package org.myorg.datastraemapi.demo;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.myorg.datastraemapi.model.Student;

import java.util.ArrayList;
import java.util.List;

public class ReduceDemo {

	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		List<Student> listSource = new ArrayList<Student>() {{
			add(new Student(1, 1));
			add(new Student(2, 2));
			add(new Student(3, 3));
			add(new Student(4, 4));
			add(new Student(5, 4));
			add(new Student(6, 4));
			add(new Student(7, 4));
		}};
		DataStreamSource<Student> dataStreamSource = env.fromCollection(listSource);


		KeyedStream<Student, Integer> keyBy = dataStreamSource.keyBy(new KeySelector<Student, Integer>() {
			@Override
			public Integer getKey(Student student) throws Exception {
				return student.age;
			}
		});

		SingleOutputStreamOperator<Student> reduce = keyBy.reduce(new ReduceFunction<Student>() {
			@Override
			public Student reduce(Student value1, Student value2) throws Exception {
				Student student1 = new Student();
				student1.id = (value1.id + value2.id) / 2;
				student1.age = (value1.age + value2.age) / 2;
				return student1;
			}
		});
		reduce.print();

		env.execute("Flink Streaming Java API Skeleton");

	}
}
